package com.dib.neo.bkfs.server;


import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.dib.neo.bkfs.net.NetworkUtils;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.InterleavedLedgerStorage;
import org.apache.bookkeeper.client.BookKeeperAdmin;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.proto.BookieServer;


import java.util.Collection;
import java.util.stream.Collectors;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;

/**
 * Utility for starting embedded Apache BookKeeper Server (Bookie)
 */
public class EmbeddedBookie implements AutoCloseable {

  private final static Logger LOG = Logger.getLogger(EmbeddedBookie.class.getName());
  private final BKFSConfiguration configuration;
  private final Path baseDirectory;
  private final List<BookieServer> bookieServers = new ArrayList<>(3);

  private final StatsLogger statsLogger;

  public EmbeddedBookie(BKFSConfiguration configuration) {
    this(configuration, null);
  }

  public EmbeddedBookie(BKFSConfiguration configuration, StatsLogger statsLogger) {
    this.configuration = configuration;
    this.baseDirectory = Paths.get(configuration
        .getString(BKFSConfiguration.PROPERTY_BASEDIR, BKFSConfiguration.PROPERTY_BASEDIR_DEFAULT));
    this.statsLogger = statsLogger != null ? statsLogger : new NullStatsLogger();
  }

  public void start() throws Exception {
    long _start = System.currentTimeMillis();
    String mode = configuration
        .getString(BKFSConfiguration.PROPERTY_MODE, BKFSConfiguration.PROPERTY_MODE_STANDALONE);
    startBookie(0);

    if (!BKFSConfiguration.PROPERTY_MODE_CLUSTER.equals(mode)) {
      int num = configuration.getInt(BKFSConfiguration.PROPERTY_BOOKKEEPER_ENSEMBLE,
          BKFSConfiguration.PROPERTY_BOOKKEEPER_ENSEMBLE_DEFAULT);
      for (int i = 1; i < num; i++) {
        startBookie(i);
      }
    }

    for (BookieServer bookieServer : bookieServers) {
      for (int i = 0; i < 100; i++) {
        if (bookieServer.getBookie().isRunning()) {
          LOG.info("Apache Bookkeeper started.");
          break;
        }
        Thread.sleep(500);
      }
    }
    long _stop = System.currentTimeMillis();
    LOG.severe("Booting Apache Bookkeeper finished. Time " + (_stop - _start) + " ms");

    //*/
  }

  private void startBookie(int bookId) throws Exception {
    ServerConfiguration conf = new ServerConfiguration();
    conf.setZkTimeout(configuration.getInt(BKFSConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT,
        BKFSConfiguration.PROPERTY_ZOOKEEPER_SESSIONTIMEOUT_DEFAULT));
    conf.setZkServers(configuration.getString(BKFSConfiguration.PROPERTY_ZOOKEEPER_ADDRESS,
        BKFSConfiguration.PROPERTY_ZOOKEEPER_ADDRESS_DEFAULT));
    conf.setStatisticsEnabled(true);
    int port = configuration.getInt(BKFSConfiguration.PROPERTY_BOOKKEEPER_BOOKIE_PORT,
        BKFSConfiguration.PROPERTY_BOOKKEEPER_BOOKIE_PORT_DEFAULT) + bookId;

    conf.setUseHostNameAsBookieID(false);
    Path bookie_dir = baseDirectory.resolve("bookie" + bookId);
    if (port <= 0) {
      Integer _port = readLocalBookiePort(bookie_dir);
      if (_port == null) {
        _port = NetworkUtils.assignFirstFreePort();
        LOG.log(Level.SEVERE, "As configuration parameter "
                + BKFSConfiguration.PROPERTY_BOOKKEEPER_BOOKIE_PORT
                + " is {0},I have choosen to listen on port {1}."
                + " Set to a positive number in order to use a fixed port",
            new Object[]{Integer.toString(port), Integer.toString(_port)});
        persistLocalBookiePort(bookie_dir, _port);
      }
      port = _port;
    }
    conf.setBookiePort(port);
    Files.createDirectories(bookie_dir);
    Path bookie_data_dir = bookie_dir.resolve("bookie_data").toAbsolutePath();
    Path bookie_journal_dir = bookie_dir.resolve("bookie_journal").toAbsolutePath();
    Files.createDirectories(bookie_data_dir);
    Files.createDirectories(bookie_journal_dir);
    conf.setLedgerDirNames(new String[]{bookie_data_dir.toString()});
    conf.setJournalDirName(bookie_journal_dir.toString());

    conf.setFlushInterval(1000);
    conf.setMaxBackupJournals(3);
    conf.setMaxJournalSizeMB(256L);
    conf.setNumAddWorkerThreads(8);
    conf.setMaxPendingReadRequestPerThread(200000); // new in 4.6
    conf.setMaxPendingAddRequestPerThread(200000); // new in 4.6
    conf.setEnableLocalTransport(true);
    conf.setJournalMaxGroupWaitMSec(10L);// default 200ms
    conf.setJournalFlushWhenQueueEmpty(true);
    conf.setAutoRecoveryDaemonEnabled(false);
    conf.setAllowLoopback(true);

    //conf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName());
    conf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class);
    conf.setGcWaitTime(1000);
    for (String key : configuration.keys()) {
      if (key.startsWith("bookie.")) {
        String bookieConf = key.substring("bookie.".length());
        String value = configuration.getString(key, null);
        conf.addProperty(bookieConf, value);
        LOG.log(Level.CONFIG, "config {0} remapped to {1}={2}",
            new Object[]{key, bookieConf, value});
      }
    }

    LOG.severe("Booting Apache Bookkeeper on port " + port);

    Files.createDirectories(bookie_dir);
    dumpBookieConfiguration(bookie_dir, conf);

    boolean forcemetaformat = configuration.getBoolean("bookie.forcemetaformat", false);
    LOG.log(Level.CONFIG, "bookie.forcemetaformat={0}", forcemetaformat);

    boolean result = BookKeeperAdmin.format(conf, false, forcemetaformat);
    if (result) {
      LOG.info("BookKeeperAdmin.format: created a new workspace on ZK");
    } else {
      LOG.info("BookKeeperAdmin.format: ZK space does not need an format operation");
    }

    boolean forceformat = configuration.getBoolean("bookie.forceformat", false);
    LOG.log(Level.CONFIG, "bookie.forceformat={0}", forceformat);
    if (forceformat) {
      result = Bookie.format(conf, false, forceformat);
      if (result) {
        LOG.info("Bookie.format: formatter applied to local bookie");
      } else {
        LOG.info("Bookie.format: local boookie did not need formatting");
      }
    }

    BookieServer bookieServer = new BookieServer(conf, statsLogger);
    bookieServers.add(bookieServer);
    bookieServer.start();

  }

  private void dumpBookieConfiguration(Path bookie_dir,
      org.apache.bookkeeper.conf.ServerConfiguration conf) throws IOException {
    // dump actual BookKeeper configuration in order to use bookkeeper shell
    Path actual_bookkeeper_configuration = bookie_dir.resolve("embedded.bookie.properties");
    StringBuilder builder = new StringBuilder();
    for (Iterator<String> key_it = conf.getKeys(); key_it.hasNext(); ) {
      String key = key_it.next() + "";
      Object value = conf.getProperty(key);
      if (value instanceof Collection) {
        value = ((Collection) value).stream().map(String::valueOf).collect(Collectors.joining(","));
      }
      builder.append(key + "=" + value + "\n");
    }
    Files
        .write(actual_bookkeeper_configuration, builder.toString().getBytes(StandardCharsets.UTF_8),
            StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE);
    LOG.severe(
        "Dumped actual Bookie configuration to " + actual_bookkeeper_configuration
            .toAbsolutePath());
  }

  @Override
  public void close() {
    for (BookieServer bookieServer : bookieServers) {
      if (bookieServer != null) {
        LOG.info("Apache Bookkeeper stopping");
        try {
          bookieServer.shutdown();

          bookieServer.join();
        } catch (InterruptedException err) {
          Thread.currentThread().interrupt();
        } finally {
          bookieServer = null;
        }
      }
    }
    bookieServers.clear();

  }

  public Integer readLocalBookiePort(Path dataPath) throws IOException {
    Path file = dataPath.resolve("bookie_port");
    try {
      LOG.log(Level.SEVERE, "Looking for local port into file {0}", file);
      if (!Files.isRegularFile(file)) {
        LOG.log(Level.SEVERE, "Cannot find file {0}", file);
        return null;
      }
      List<String> lines = Files.readAllLines(file, StandardCharsets.UTF_8);
      for (String line : lines) {
        line = line.trim().toLowerCase();
        // skip comments and empty lines
        if (line.startsWith("#") || line.isEmpty()) {
          continue;
        }
        int res = Integer.parseInt(line);
        LOG.log(Level.SEVERE, "Found local port {0} into file {1}",
            new Object[]{Integer.toString(res), file});
        return res;
      }
      throw new IOException("Cannot find any valid line inside file " + file.toAbsolutePath());
    } catch (IOException error) {
      LOG.log(Level.SEVERE, "Error while reading file " + file.toAbsolutePath(), error);
      throw error;
    }
  }

  public void persistLocalBookiePort(Path dataPath, int port) throws IOException {
    Files.createDirectories(dataPath);
    Path file = dataPath.resolve("bookie_port");
    StringBuilder message = new StringBuilder();
    message.append("# This file contains the port of the bookie used by this node\n");
    message.append(
        "# Do not change the contents of this file, otherwise the beheaviour of the system will\n");
    message.append("# lead eventually to data loss\n");
    message.append("# \n");
    message.append("# Any line which starts with '#' and and blank line will be ignored\n");
    message.append("# The system will consider the first non-blank line as port\n");
    message.append("\n\n");
    message.append(port);
    Files.write(file, message.toString().getBytes(StandardCharsets.UTF_8),
        StandardOpenOption.CREATE_NEW);
  }
}
